package com.gzhuxj201.service.Impl;

import cn.hutool.core.bean.BeanUtil;
import com.gzhuxj201.entity.Course;
import com.gzhuxj201.entity.Result;
import com.gzhuxj201.entity.SC;
import com.gzhuxj201.mapper.CourseMapper;
import com.gzhuxj201.mapper.SCMapper;
import com.gzhuxj201.service.SCService;
import com.gzhuxj201.util.RedisConstants;
import lombok.extern.slf4j.Slf4j;
import org.apache.poi.ss.usermodel.*;
import org.apache.poi.xssf.usermodel.XSSFWorkbook;
import org.redisson.api.RLock;
import org.redisson.api.RedissonClient;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.core.io.ClassPathResource;
import org.springframework.data.redis.connection.stream.*;
import org.springframework.data.redis.core.StringRedisTemplate;
import org.springframework.data.redis.core.script.DefaultRedisScript;
import org.springframework.scheduling.concurrent.ThreadPoolExecutorFactoryBean;
import org.springframework.stereotype.Service;
import org.springframework.transaction.annotation.Transactional;
import org.springframework.web.multipart.MultipartFile;

import javax.annotation.PostConstruct;
import java.io.FileInputStream;
import java.io.IOException;
import java.io.PrintWriter;
import java.time.Duration;
import java.util.Collections;
import java.util.Date;
import java.util.List;
import java.util.Map;
import java.util.concurrent.*;

@Service
@Slf4j
public class SCServiceImpl implements SCService {
    @Autowired
    private SCMapper scMapper;
    @Autowired
    private CourseMapper courseMapper;
    @Autowired
    private StringRedisTemplate stringRedisTemplate;
    @Autowired
    private RedissonClient redissonClient;
/*    @Autowired
    private SCService scService;*/

    //加载lua脚本文件资源
    private static final DefaultRedisScript<Long> COURSE_SCRIPT;
    static {
        COURSE_SCRIPT = new DefaultRedisScript<>();
        COURSE_SCRIPT.setLocation(new ClassPathResource("courseSelect.lua"));
        COURSE_SCRIPT.setResultType(Long.class);
    }
    private static final DefaultRedisScript<Long> COURSE_DESELECT_SCRIPT;
    static {
        COURSE_DESELECT_SCRIPT = new DefaultRedisScript<>();
        COURSE_DESELECT_SCRIPT.setLocation(new ClassPathResource("courseDeselect.lua"));
        COURSE_DESELECT_SCRIPT.setResultType(Long.class);
    }

    //开线程池来处理选课/退课的业务
  /*  private static final ExecutorService COURSE_SELECTION_EXECUTOR = Executors.newSingleThreadExecutor();
    private static final ExecutorService COURSE_DESELECTION_EXECUTOR = Executors.newSingleThreadExecutor();*/
    //private  static final  ExecutorService COURSE_EXECUTOR = Executors.newFixedThreadPool(2);
    ExecutorService a=Executors.newScheduledThreadPool(5);
    private  static final ThreadPoolExecutor COURSE_EXECUTOR = new ThreadPoolExecutor(
            2, 2, 0, TimeUnit.SECONDS,
            new LinkedBlockingDeque(2),
            r -> new Thread(r),
            new ThreadPoolExecutor.AbortPolicy()
    );

    @PostConstruct //在项目启动之后就用线程池来处理消息队列中的业务
    private void init() {
        //可以加一个是否在选课时间内,在选课时间内才处理
        /*COURSE_SELECTION_EXECUTOR.submit(new CourseSelectHandle());
        COURSE_DESELECTION_EXECUTOR.submit(new CourseDeselectHandle());*/
        COURSE_EXECUTOR.submit(new CourseDeselectHandle());
        COURSE_EXECUTOR.submit(new CourseSelectHandle());
    }


    private class CourseDeselectHandle implements Runnable {

        @Override
        public void run() {
           // stringRedisTemplate.opsForStream().destroyGroup("stream.courseDeselections","dcg1");
            try {//创建消费者组
                stringRedisTemplate.opsForStream().createGroup("stream.courseDeselections",ReadOffset.from("$"),"dcg1");
                System.err.println("退课:消费者组创建成功!");
            } catch (Exception e) {
                System.err.println("退课:消费者组创建失败!");
            }
            while (true) {
                try {
                    // 1.获取消息队列中的选课信息 XREADGROUP GROUP dcg1 c1 COUNT 1 BLOCK 2000 STREAMS stream.courseDeselections >
                    // > :表示读取队列中下一个未处理的消息  $ : 表示读取最新消息  0: 表示从pending-list中读取未确定的信息
                    List<MapRecord<String, Object, Object>> list =  stringRedisTemplate.opsForStream().read(
                            Consumer.from("dcg1","c1")
                            , StreamReadOptions.empty().count(1).block(Duration.ofSeconds(2L)),
                            StreamOffset.create("stream.courseDeselections", ReadOffset.from(">"))
                    );

                    // 2.判断选课信息是否为空
                    if (list == null || list.isEmpty()) {
                        // 如果为null，说明没有消息，继续下一次循环
                        continue;
                    }
                    // 解析数据 map中 string为消息id 另外两个object为键值对
                    MapRecord<String, Object, Object> record = list.get(0);
                    Map<Object, Object> value = record.getValue();
                    SC sc = BeanUtil.fillBeanWithMap(value, new SC(), true);
                    // 3.执行退课
                    deselectByCnoHandle(sc);
                    // 4.确认消息 XACK
                    stringRedisTemplate.opsForStream().acknowledge("stream.courseDeselections", "dcg1", record.getId());
                } catch (Exception e) {
                    log.error("处理选课异常", e);

                    try {//创建消费者组
                        stringRedisTemplate.opsForStream().createGroup("stream.courseDeselections",ReadOffset.from("$"),"dcg1");
                    } catch (Exception exception) {
                    }
                    handlePendingList();
                }
            }
        }

        private void handlePendingList() {
            while (true) {
                try {
                    // 1.获取pending-list中的选课信息 XREADGROUP GROUP cg1 c1 COUNT 1 BLOCK 2000 STREAMS stream.courseDeselections 0
                    // 0: 表示从pending-list中读取未确定的信息

                    List<MapRecord<String, Object, Object>> list = stringRedisTemplate.opsForStream().read(
                            Consumer.from("dcg1", "c1"),
                            StreamReadOptions.empty().count(1),
                            StreamOffset.create("stream.courseDeselections", ReadOffset.from("0"))
                    );
                    // 2.判断选课信息是否为空
                    if (list == null || list.isEmpty()) {
                        // 如果为null，说明没有异常消息，结束循环
                        break;
                    }
                    // 解析数据
                    MapRecord<String, Object, Object> record = list.get(0);
                    Map<Object, Object> value = record.getValue();
                    SC sc = BeanUtil.fillBeanWithMap(value, new SC(), true);
                    // 3.执行退课
                    deselectByCnoHandle(sc);
                    // 4.确认消息 XACK
                    stringRedisTemplate.opsForStream().acknowledge("stream.courseDeselections", "dcg1", record.getId());
                } catch (Exception e) {
                    log.error("处理选课异常", e);
                }
            }
        }
    }

    private class CourseSelectHandle implements Runnable {

        @Override
        public void run() {
            try {//创建消费者组
                stringRedisTemplate.opsForStream().createGroup("stream.courseSelections",ReadOffset.from("$"),"cg1");
                System.err.println("选课:消费者组创建成功!");
            } catch (Exception e) {
                System.err.println("选课:消费者组创建失败!");
            }
            while (true) {
                try {
                    // 1.获取消息队列中的选课信息 XREADGROUP GROUP cg1 c1 COUNT 1 BLOCK 2000 STREAMS stream.courseSelections >
                    // > :表示读取队列中下一个未处理的消息  $ : 表示读取最新消息  0: 表示从pending-list中读取未确定的信息
                    List<MapRecord<String, Object, Object>> list = stringRedisTemplate.opsForStream().read(
                            Consumer.from("cg1", "c1"),
                            StreamReadOptions.empty().count(1).block(Duration.ofSeconds(2)),
                            StreamOffset.create("stream.courseSelections", ReadOffset.lastConsumed())
                    );
                    // 2.判断选课信息是否为空
                    if (list == null || list.isEmpty()) {
                        // 如果为null，说明没有消息，继续下一次循环
                        continue;
                    }
                    // 解析数据 map中 string为消息id 另外两个object为键值对
                    MapRecord<String, Object, Object> record = list.get(0);
                    Map<Object, Object> value = record.getValue();
                    SC sc = BeanUtil.fillBeanWithMap(value, new SC(), true);
                    // 3.创建选课
                    addHandle(sc);
                    // 4.确认消息 XACK
                    stringRedisTemplate.opsForStream().acknowledge("stream.courseSelections", "cg1", record.getId());
                } catch (Exception e) {

                    log.error("处理选课异常", e);
                    try {//创建消费者组
                        stringRedisTemplate.opsForStream().createGroup("stream.courseSelections",ReadOffset.from("$"),"cg1");
                    } catch (Exception exception) {
                    }

                    handlePendingList();
                }
            }
        }

        private void handlePendingList() {
            while (true) {
                try {
                    // 1.获取pending-list中的选课信息 XREADGROUP GROUP cg1 c1 COUNT 1 BLOCK 2000 STREAMS stream.courseSelections 0
                    // 0: 表示从pending-list中读取未确定的信息
                    List<MapRecord<String, Object, Object>> list = stringRedisTemplate.opsForStream().read(
                            Consumer.from("cg1", "c1"),
                            StreamReadOptions.empty().count(1),
                            StreamOffset.create("stream.courseSelections", ReadOffset.from("0"))
                    );
                    // 2.判断选课信息是否为空
                    if (list == null || list.isEmpty()) {
                        // 如果为null，说明没有异常消息，结束循环
                        break;
                    }
                    // 解析数据
                    MapRecord<String, Object, Object> record = list.get(0);
                    Map<Object, Object> value = record.getValue();
                    SC sc = BeanUtil.fillBeanWithMap(value, new SC(), true);
                    // 3.创建选课
                    addHandle(sc);
                    // 4.确认消息 XACK
                    stringRedisTemplate.opsForStream().acknowledge("stream.courseSelections", "cg1", record.getId());
                } catch (Exception e) {
                    log.error("处理选课异常", e);
                }
            }
        }
    }


    @Override
    public List<String> selectBySno(String sno) {
        return scMapper.selectBySno(sno);
    }

    @Override
    public Result add(SC sc) {
        String cno = sc.getCno();
        String sno = sc.getSno();


        // 1.执行lua脚本
        Long result = stringRedisTemplate.execute(
                COURSE_SCRIPT,
                Collections.emptyList(),
                cno, sno
        );
        int r = result.intValue();
        // 2.判断结果是否为0
        if (r != 0) {
            // 2.1.不为0 ，代表没有选课资格
            return Result.error(r == 1 ? "库存不足" : "不能重复选课");
        }
        // 3.返回课程号
        return Result.success(cno);
    }

    @Override
    public Result deselectByCno(SC sc) {
        String cno = sc.getCno();
        String sno = sc.getSno();
        String key = RedisConstants.COURSE_INITNUM_KEY + cno;
        String initStock = stringRedisTemplate.opsForValue().get(key);


        // 1.执行lua脚本
        Long result = stringRedisTemplate.execute(
                COURSE_DESELECT_SCRIPT,
                Collections.emptyList(),
                cno, sno ,initStock
        );
        int r = result.intValue();
        // 2.判断结果是否为0
        if (r != 0) {
            // 2.1.不为0 ，代表没有退课资格
            return Result.error(r == 1 ? "退课超出上限" : "不能重复退课");
        }
        // 3.返回课程号
        return Result.success(cno);
    }



    @Transactional
    public Result addHandle(SC sc) {
        String sno = sc.getSno();
        String cno = sc.getCno();
        // 1.一人一选
        // 创建锁对象
        RLock redisLock =redissonClient.getLock("lock:selectCourse:" + sno);
        // 尝试获取锁
        boolean isLock = redisLock.tryLock();
        // 判断
        if (!isLock) {
            // 获取锁失败，直接返回失败或者重试
            log.error("不允许重复选课！");
            return null;
        }

        try {
            // 2.查询选课表
            SC sc1 = scMapper.selectBySnoAndCno(sno,cno);
            // 2.1.判断是否已选课
            if (sc1 != null) {
                //学生已选上了这门课
                return Result.error("请勿重复选课!");
            }

            //乐观锁解决超选问题
            // 3.扣减库存  扣减选课数量
            int res = courseMapper.updateCurNumByCno(cno);
            if (res == 0) {
                // 扣减失败
                return Result.error("该课程已选满");
            }

            //4.保存选课信息到sc表
            int res2 = scMapper.add(new SC(sno, cno));

            // 5.返回课程号
            return Result.success(cno);
        } finally {
            redisLock.unlock();
        }

    }



    @Transactional
    public Result deselectByCnoHandle(SC sc) {
        String sno = sc.getSno();
        String cno = sc.getCno();
        // 创建锁对象
        RLock redisLock =redissonClient.getLock("lock:DeselectCourse:" + sno);
        // 尝试获取锁
        boolean isLock = redisLock.tryLock();
        // 判断
        if (!isLock) {
            // 获取锁失败，直接返回失败或者重试
            log.error("不允许重复选课！");
            return null;
        }

        try {
            // 5.1.查询选课表
            SC sc1 = scMapper.selectBySnoAndCno(sno,cno);
            // 5.2.判断是否已选课
            if (sc1 == null) {
                //学生已退了这门课
                return Result.error("请勿重复退课!");
            }

            //乐观锁解决超退问题
            // 6.退课,增加本课程可选人数
            Course course = courseMapper.selectByCno(cno);//获取当前选课人数

            int res = courseMapper.updateCurNumByCourse(course);//比对选课人数,如果不一样,则返回请重试
            if (res == 0) {
                // 扣减失败
                return Result.error("退课异常,请重试");
            }

            //删除sc表选课信息
            int res2 = scMapper.deleteBySnoAndCno(sc);

            // 7.返回课程号
            return Result.success(cno);
        } finally {
            redisLock.unlock();
        }

    }

    @Override
    @Transactional
    public void importStuToCourse(MultipartFile file, String cno) throws IOException {
        Workbook workbook = new XSSFWorkbook(file.getInputStream()); //根据上传的输入流创建workbook
        //Workbook workbook = new XSSFWorkbook(new FileInputStream("D://学生数据.xlsx")); //根据上传的输入流创建workbook

        Sheet sheet = workbook.getSheetAt(0); //获取工作薄中的第一个工作表
        int lastRowIndex = sheet.getLastRowNum(); //获取这个sheet中最后一行数据，为了循环遍历

        //        以下三个为了节省栈内存，所以提到循环的外面
        Row row = null;
        Cell cell = null;

        //开始循环每行，获取每行的单元格中的值，放入到user属性中
        for (int i = 1; i <= lastRowIndex; i++) {
            row = sheet.getRow(i);
            SC sc= new SC();
            //因为第一个列单元格中是字符串，可以直接使用getStringCellValue方法
            row.getCell(0).setCellType(CellType.STRING);
            String sno = row.getCell(0).getStringCellValue();
            //设置学号
            sc.setSno(sno);
            //设置课程
            sc.setCno(cno);
            //插入数据库
            scMapper.add(sc);
        }

    }


/*    @Override
    //@Transactional
    public Result add(SC sc) {
        //悲观锁解决一人一选问题  intern()把字符串变为常量,
        //如不用intern()字符串每层都会单独new一个新的对象,所以每层的锁是不一样的,不能解决每个用户并发请求串行处理
        synchronized (sc.getSno().toString().intern()) {
            //通过this.addHandle(sc)调用会导致事务失效  方法不加public也会导致事务失效
            //解决
            //方法一: spring管理的事务,要用自己的代理对象来调用,否则会导致事务失效
            //条件:添加 aspectjweaver 依赖; 设置@EnableAspectJAutoProxy(exposeProxy = true)暴露代理对象的属性
            SCService proxy = (SCService) AopContext.currentProxy();
            return proxy.addHandle(sc);
            //方法二: 通过自己service的接口来调用
            //return scService.addHandle(sc);
            //方法三: 在本方法上再加一个事务
            //return addHandle(sc);
        }

    }*/

/*    @Override
    public Result deselectByCno(SC sc) {
        synchronized (sc.getSno().toString().intern()) {
            //方法一: spring管理的事务,要用自己的代理对象来调用,否则会导致事务失效
            //条件:添加 aspectjweaver 依赖; 设置@EnableAspectJAutoProxy(exposeProxy = true)暴露代理对象的属性
            SCService proxy = (SCService) AopContext.currentProxy();
            return proxy.deselectByCnoHandle(sc);
        }
    }*/
}



